/**
 * Copyright (c) Huawei Technologies Co., Ltd. 2023-2023. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "framework/aicpu_rpc_serverv2.h"
#include <numeric>
#include "common/aicpu_hccl_common.h"
#include "utils/mc2_aicpu_utils.h"
#include "hccl_common.h"
#include "framework/aicpu_hccl_process.h"
#include "securec.h"

static constexpr uint32_t AC_MSG_VALID_MASK_TAG = 0x5CDF123A;
static constexpr uint16_t TURN_LEFT_SHIFT_BIT = 16;

HcclResult AicpuRpcServerV2::Init(const HcclMC2WorkSpace &workspaceInfo)
{
    // 为提升效率，workspace 必须512 对齐
    u64 addr = workspaceInfo.workSpace;
    if (addr & 0x1ff) {
        addr = (addr & (~((uint64_t)0x1ff))) + 0x200;
    }
    HCCL_INFO("Align hcclmsgarea from %p to %p", workspaceInfo.workSpace, addr);
    hcclMsgArea_ = reinterpret_cast<HcclMsgArea *>(addr);
    msgPos_ = 0;
    for (int8_t i = 0; i < MC2_HCCL_MAX_HANDLE_ID; i++) {
        handleIdMsgPosition_[i] = -1;
    }
    isFinalize_ = false;
    prepareTime_ = GetCurCpuTimestamp();
    eventPrintTurn_ = 1;               // 标记 Event日志的打印，20s之后打印一条 event日志
    turnNumAddr_ = reinterpret_cast<u64>(hcclMsgArea_) + sizeof(HcclMsgArea);
    if (turnNumAddr_ + sizeof(u32) * TILING_TURN_MAX * AC_MAX_RANK_NUM_V2 >
        workspaceInfo.workSpace + workspaceInfo.workSpaceSize) {
        HCCL_ERROR("Turn number addr %#llx, space for turn number is %lu, the space after workspace %#llx will "
            "be overwrited.", turnNumAddr_, sizeof(u32) * TILING_TURN_MAX * AC_MAX_RANK_NUM_V2,
            workspaceInfo.workSpace + workspaceInfo.workSpaceSize);
        return HCCL_E_INTERNAL;
    }
    uint32_t *turnNums = reinterpret_cast<uint32_t *>(turnNumAddr_);
    std::iota(&turnNums[0], &turnNums[TILING_TURN_MAX * AC_MAX_RANK_NUM_V2], 0);
    return HCCL_SUCCESS;
}

void AicpuRpcServerV2::Reset()
{
    msgPos_ = 0;
    msgPosForKernel_ = 0;
    for (int8_t i = 0; i < MC2_HCCL_MAX_HANDLE_ID; i++) {
        handleIdMsgPosition_[i] = -1;
    }
    isFinalize_ = false;
    prepareTime_ = GetCurCpuTimestamp();
    eventPrintTurn_ = 1;
}

void AicpuRpcServerV2::SetMsgHandlePos(uint32_t msgPos, HcclHandle handleId) {
    if (handleId >= MC2_HCCL_MAX_HANDLE_ID || handleId < 0) {
        return;
    }
    handleIdMsgPosition_[handleId] = msgPos;
}

int32_t AicpuRpcServerV2::GetMsgHandlePos(HcclHandle handleId) {
    if (handleId >= MC2_HCCL_MAX_HANDLE_ID || handleId < 0) {
        HCCL_ERROR("[GetMsgHandlePos] invalid handleId %d", handleId);
        return -1;
    }
    if (handleIdMsgPosition_[handleId] < 0) {
        HCCL_WARNING("[GetMsgHandlePos] invalid handleIdMsgPosition %d", handleIdMsgPosition_[handleId]);
        return -1;
    }
    return handleIdMsgPosition_[handleId];
}

uint32_t AicpuRpcServerV2::GetMsgPos(void) const {
    return msgPos_;
}

void AicpuRpcServerV2::SetMsgPos(uint32_t pos) {
    msgPos_ = pos;
}

void AicpuRpcServerV2::SetIsFinalize(bool finalize) {
    isFinalize_ = finalize;
}

bool AicpuRpcServerV2::GetIsFinalize(void) const {
    return isFinalize_;
}

HcclMsgExt* AicpuRpcServerV2::GetHcclMsgExtPtr() {
    return msgExt_.get();
}

HcclMsgArea* AicpuRpcServerV2::GetHcclMsgArea(void) {
    return hcclMsgArea_;
}

HcclMsg *AicpuRpcServerV2::GetMsgWorkSpace(void) {
    return (HcclMsg *)(&hcclMsgArea_->sendMsgList[0]);
}
uint64_t AicpuRpcServerV2::GetWorkSpaceCommitArea(void) {
    return (uint64_t)(reinterpret_cast<uint64_t>(hcclMsgArea_) + offsetof(HcclMsgArea, commitTurnCnt));
}

uint64_t AicpuRpcServerV2::GetWorkSpaceFinishArea(void) {
    return (uint64_t)(reinterpret_cast<uint64_t>(hcclMsgArea_) + offsetof(HcclMsgArea, finishedTurnCnt));
}

uint64_t AicpuRpcServerV2::GetFinishAddr(int32_t idx) {
    if (idx >= static_cast<int32_t>(AC_MSG_CNT)) {
        HCCL_ERROR("GetFinishAddr idx is %d exceed AC_MSG_CNT", idx);
        return 0;
    }
    TurnCnt *curCnt = reinterpret_cast<TurnCnt *>((reinterpret_cast<uint64_t>(hcclMsgArea_) +
                                                   offsetof(HcclMsgArea, finishedTurnCnt) + idx * sizeof(TurnCnt)));
    return (uint64_t)reinterpret_cast<uint64_t>(&curCnt->cnt);
}

uint64_t AicpuRpcServerV2::GenXorForMsgExt(int32_t idx, u32 rankSize) {
    uint64_t xorVal = 0U;
    for (uint32_t i = 0U; i < rankSize; ++i) {
        xorVal ^= hcclMsgArea_->paramExtMsgList[idx].sendCounts[i];
        xorVal ^= hcclMsgArea_->paramExtMsgList[idx].sendOffset[i];
        xorVal ^= hcclMsgArea_->paramExtMsgList[idx].recvCounts[i];
        xorVal ^= hcclMsgArea_->paramExtMsgList[idx].recvOffset[i];
    }
    xorVal ^= hcclMsgArea_->paramExtMsgList[idx].valid;
    return xorVal;
}

uint64_t AicpuRpcServerV2::GetCommitareaAddr(int32_t idx) {
    if (idx >= static_cast<int32_t>(AC_MSG_CNT)) {
        HCCL_ERROR("GetCommitareaAddr idx is %d exceed AC_MSG_CNT", idx);
        return 0;
    }
    TurnCnt *curCnt = reinterpret_cast<TurnCnt *>((reinterpret_cast<uint64_t>(hcclMsgArea_) +
                                                   offsetof(HcclMsgArea, commitTurnCnt) + idx * sizeof(TurnCnt)));
    return (uint64_t)reinterpret_cast<uint64_t>(&curCnt->cnt);
}

HcclResult AicpuRpcServerV2::AddCcoreWait(uint16_t streamId, u64 waitAddr, uint32_t turnNum, hccl::Stream *stream, bool isLast) // client commit wait
{
    (void)streamId;
    uint8_t *sqeBuffer = nullptr;
    uint8_t *sqeTypeAddr = nullptr;
    uint8_t *sqeDfxInfoAddr = nullptr;
    uint16_t taskId = 0U;
    CHK_RET(stream->GetNextSqeBufferAddr(sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
    const HcclComStreamInfo &streamInfo = stream->GetHcclStreamInfo();
    AddOneWaitStartSqe(streamInfo.actualStreamId, taskId, waitAddr, turnNumAddr_ + turnNum * sizeof(u32),
        isLast, reinterpret_cast<rtStarsCcoreWaitStartSqe_t *>(sqeBuffer), sqeTypeAddr);
    hccl::HcclSqeContext* sqeCtx = stream->GetSqeContextPtr();
    if (sqeCtx == nullptr) {
        HCCL_ERROR("AddCcoreWait sqeCtx is nullptr");
        return HCCL_E_INTERNAL;
    }
    sqeCtx->buffer.addInfo[taskId % hccl::HCCL_SQE_MAX_CNT] =
        ((turnNum << TURN_LEFT_SHIFT_BIT) + static_cast<uint32_t>(isLast));
    return HCCL_SUCCESS;
}

HcclResult AicpuRpcServerV2::AddCcoreNotify(uint16_t streamId, u64 recordAddr, uint32_t turnNum, hccl::Stream *stream) // client finish notify
{
    (void)streamId;
    uint8_t *sqeBuffer = nullptr;
    uint8_t *sqeTypeAddr = nullptr;
    uint8_t *sqeDfxInfoAddr = nullptr;
    uint16_t taskId = 0U;
    CHK_RET(stream->GetNextSqeBufferAddr(sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
    const HcclComStreamInfo &streamInfo = stream->GetHcclStreamInfo();
    AddOneWriteValueStartSqe(streamInfo.actualStreamId, taskId, recordAddr,
        turnNumAddr_ + turnNum * sizeof(u32), reinterpret_cast<rtStarsCcoreWriteValueSqe_t *>(sqeBuffer),
        sqeTypeAddr);
    hccl::HcclSqeContext* sqeCtx = stream->GetSqeContextPtr();
    if (sqeCtx == nullptr) {
        HCCL_ERROR("AddCcoreNotify sqeCtx is nullptr");
        return HCCL_E_INTERNAL;
    }
    sqeCtx->buffer.addInfo[taskId % hccl::HCCL_SQE_MAX_CNT] = turnNum;
    return HCCL_SUCCESS;
}

uint64_t AicpuRpcServerV2::GetFinishAddrByHandleId(HcclHandle handleId)
{
    int32_t msgPos = GetMsgHandlePos(handleId);
    if (msgPos < 0) {
        return 0;
    }
    return GetFinishAddr(msgPos);
}

void AicpuRpcServerV2::SetMsgRepeatCnt(u8 repeatCnt)
{
    if (curStepSize_ == 0U) {
        repeatCnt_[msgPos_] = repeatCnt;
    } else {
        repeatCnt_[msgPos_] = repeatCnt * curStepSize_;
    }
}

int32_t AicpuRpcServerV2::GetMsgRepeatCnt(HcclHandle handleId)
{
    int32_t msgPos = GetMsgHandlePos(handleId);
    if (msgPos < 0) {
        return -1;
    }
    return repeatCnt_[msgPos];
}

HcclResult AicpuRpcServerV2::ProcessExpectPrepareMsg(CommonHcclMsg *hcclMsg, uint8_t expectId)
{
    // 当前无翻转场景，只需考虑单个通信域提前Finializa场景
    if ((hcclMsg->seqNum == 0) && (expectId > 0)) {
        return HCCL_SUCCESS;
    }
    if (hcclMsg->seqNum < expectId) {
        HCCL_ERROR("curMsg seqNum %d is smaller than expect %d ignore.", hcclMsg->seqNum, expectId);
        return HCCL_E_INTERNAL;
    }
    if (hcclMsg->seqNum > expectId) {
        HCCL_INFO("curMsg seqNum %d is bigger than expect %d ignore.", hcclMsg->seqNum, expectId);
        return HCCL_E_UNAVAIL;
    }
    return HCCL_SUCCESS;
}

void AicpuRpcServerV2::SetNeedRetryFlag(bool needRetryFlag)
{
    needReProcess_ = needRetryFlag;
}

bool AicpuRpcServerV2::ReadValidMsgExtArea(int32_t idx, u32 rankSize)
{
#ifdef __aarch64__
        __asm__ __volatile__("dsb ld" : : : "memory");
#endif
#ifdef __amd64__
        __asm__ __volatile__("" : : : "memory");
#endif
    if (hcclMsgArea_->paramExtMsgList[idx].valid != static_cast<u64>(AC_MSG_VALID_MASK_TAG)) {
        return false;
    }
    uint64_t msgExtXorCheck = GenXorForMsgExt(idx, rankSize);
    static uint32_t msgExtXorCheckTurn = 0;
    if (msgExtXorCheckTurn % MC2_API_XORCHECK_PRINT_NUM == 0 &&
        msgExtXorCheck != hcclMsgArea_->paramExtMsgList[idx].xorCheck) {
        HCCL_WARNING("data is modified! modified_xor:%llu, origin_xor:%llu.", msgExtXorCheck,
            hcclMsgArea_->paramExtMsgList[idx].xorCheck);
        msgExtXorCheckTurn++;
        return false;
    }
    HCCL_INFO("hcclMsgArea xorCheck[%llu]", hcclMsgArea_->paramExtMsgList[idx].xorCheck);
    CHK_PRT_RET(memcpy_s(msgExt_->sendCounts, sizeof(uint64_t) * rankSize,
        hcclMsgArea_->paramExtMsgList[idx].sendCounts, sizeof(uint64_t) * rankSize) != EOK,
        HCCL_WARNING("memcpy_s sendCounts failed."), false);
    CHK_PRT_RET(memcpy_s(msgExt_->sendOffset, sizeof(uint64_t) * rankSize,
        hcclMsgArea_->paramExtMsgList[idx].sendOffset, sizeof(uint64_t) * rankSize) != EOK,
        HCCL_WARNING("memcpy_s sendOffset failed."), false);
    CHK_PRT_RET(memcpy_s(msgExt_->recvCounts, sizeof(uint64_t) * rankSize,
        hcclMsgArea_->paramExtMsgList[idx].recvCounts, sizeof(uint64_t) * rankSize) != EOK,
        HCCL_WARNING("memcpy_s recvCounts failed."), false);
    CHK_PRT_RET(memcpy_s(msgExt_->recvOffset, sizeof(uint64_t) * rankSize,
        hcclMsgArea_->paramExtMsgList[idx].recvOffset, sizeof(uint64_t) * rankSize) != EOK,
        HCCL_WARNING("memcpy_s recvOffset failed."), false);
    CHK_PRT_RET(memcpy_s(msgExt_->reserved, sizeof(uint64_t) * HCCL_MSG_EXT_CHECK_LEN,
        hcclMsgArea_->paramExtMsgList[idx].reserved, sizeof(uint64_t) * HCCL_MSG_EXT_CHECK_LEN) != EOK,
        HCCL_WARNING("memcpy_s reserved failed."), false);

#ifdef __aarch64__
        __asm__ __volatile__("dsb ld" : : : "memory");
#endif
#ifdef __amd64__
        __asm__ __volatile__("" : : : "memory");
#endif
    hcclMsgArea_->paramExtMsgList[idx].valid = static_cast<u64>(~AC_MSG_VALID_MASK_TAG);
#ifdef __aarch64__
        __asm__ __volatile__("dsb st" : : : "memory");
#endif
    HCCL_INFO("reset paramExtMsgList valid value %lu", hcclMsgArea_->paramExtMsgList[idx].valid);
    return true;
}

bool AicpuRpcServerV2::ReadValidMsg(HcclMsg *rMsg, HcclMsg *msg, bool reset, bool needReProcess,
    uint32_t msgPos, u32 rankSize)
{
#ifdef __aarch64__
        __asm__ __volatile__("dsb ld" : : : "memory");
#endif
#ifdef __amd64__
        __asm__ __volatile__("" : : : "memory");
#endif
    // 重处理消息
    if (needReProcess) {
        *rMsg = *msg;
        return true;
    }
    if (msg->valid != AC_MSG_VALID_MASK_TAG) {
        return false;
    }
    memcpy_s(rMsg, sizeof(HcclMsg), msg, sizeof(HcclMsg));
    uint32_t msgXorCheck = MC2AicpuUtils::GenXor(rMsg);
    static uint32_t msgXorCheckTurn = 0;
    if (msgXorCheckTurn % MC2_API_XORCHECK_PRINT_NUM == 0 && msgXorCheck != rMsg->xorCheck) {
        rMsg->PrintMsg("ReadValidMsg rMsg");
        msg->PrintMsg("ReadValidMsg rMsg");
        HCCL_WARNING("data is modified! modified_xor:%u, origin_xor:%u.", msgXorCheck, rMsg->xorCheck);
        msgXorCheckTurn++;
        return false;
    }
    if (rMsg->commType == HCCL_CMD_ALLTOALLV && !ReadValidMsgExtArea(msgPos, rankSize)) {
        return false;
    }
#ifdef __aarch64__
        __asm__ __volatile__("dsb ld" : : : "memory");
#endif
#ifdef __amd64__
        __asm__ __volatile__("" : : : "memory");
#endif

    if (reset) {
        msg->valid = ~AC_MSG_VALID_MASK_TAG;
    }
#ifdef __aarch64__
        __asm__ __volatile__("dsb st" : : : "memory");
#endif
    HCCL_INFO("reset valid value %u", msg->valid);
    return true;
}

bool AicpuRpcServerV2::ReadAddrMsg(AicpuRpcServerV2 *rpcInst, HcclMsg *hcclMsg, HcclMsg *msgList, uint32_t msgPos, u32 rankSize)
{
    if (rpcInst == nullptr || hcclMsg == nullptr || msgList == nullptr || msgPos >= static_cast<int32_t>(AC_MSG_CNT)) {
        return false;
    }
    return rpcInst->ReadAddrMsg(hcclMsg, msgList, msgPos, rankSize);
}

bool AicpuRpcServerV2::ReadAddrMsg(HcclMsg *hcclMsg, HcclMsg *msgList, uint32_t msgPos, u32 rankSize)
{
    bool ret = ReadValidMsg(hcclMsg, &(msgList[msgPos]), true, needReProcess_, msgPos, rankSize);
#ifdef CCL_LLT
    prepareTime_ = prepareTime_ - static_cast<unsigned long long>(NSEC_PER_SEC) * MC2_API_MSG_TIMEOUT
        - static_cast<unsigned long long>(NSEC_PER_SEC);
#endif
    if (GetCurCpuTimestamp() - prepareTime_ >
        static_cast<unsigned long long>(NSEC_PER_SEC) * MC2_API_MSG_TIMEOUT * eventPrintTurn_) {
        // Prepare 等待 20s
        HCCL_RUN_INFO("[AicpuRpcServerV2][ReadAddrMsg] ReadValidMsg timeout %lus", MC2_API_MSG_TIMEOUT);
        eventPrintTurn_ *= 2;  // 2 is print event log times
    }
    if (ret) {
        // 适配 printmsg
        HCCL_DEBUG("read valid msg msgPos %u commType %u", msgPos, static_cast<uint32_t>(hcclMsg->commType));
        hcclMsg->PrintMsg("ReadAddrMsg");
        // Prepare 成功，打印耗时
        u64 prepareTime = GetCurCpuTimestamp();
        if (eventPrintTurn_ > 1) {
            HCCL_RUN_INFO("[AicpuRpcServerV2][ReadAddrMsg] Read HcclMsg cost %llu", prepareTime - prepareTime_);
        } else {
            HCCL_INFO("[AicpuRpcServerV2][ReadAddrMsg] Read HcclMsg cost %llu", prepareTime - prepareTime_);
        }
        prepareTime_ = prepareTime;
        eventPrintTurn_ = 1;
        return true;
    }
    return false;
}

// reset消息区msgPos的commitTurnId
HcclResult AicpuRpcServerV2::ResetCommitTaskAdd(u16 streamId, hccl::Stream *stream)
{
    (void)streamId;
    // reset函数复用 AddCcoreWait,turnNum保证条件算子恒成立
    for (uint32_t i = 0; i < static_cast<uint32_t>(msgPos_); i++) {
        uint64_t waitAddr = GetCommitareaAddr(i);
        CHK_RET(AddCcoreWait(0, waitAddr, 0, stream, true));
    }
    return HCCL_SUCCESS;
}

void AicpuRpcServerV2::SetMsgPosForKernel(uint32_t msgPos)
{
    msgPosForKernel_ = msgPos;
}
uint32_t AicpuRpcServerV2::GetMsgPosForKernel(void)
{
    return msgPosForKernel_;
}

void AicpuRpcServerV2::WriteFinishWhenAllFinalize()
{
    uint32_t msgPos = GetMsgPos();
    hcclMsgArea_->finishedTurnCnt[msgPos].cnt = MC2_API_TurnCnt_VALID;  // 1234567899999999999 用于校验的非法值
    HCCL_INFO("Post finishedTurnCnt[%u].cnt = %lu.", msgPos, hcclMsgArea_->finishedTurnCnt[msgPos].cnt);
    #ifdef __aarch64__
    __asm__ __volatile__("dsb st" : : : "memory");
    #endif
}

void AicpuRpcServerV2::WriteRestartFlag() 
{
    for (uint32_t i = 0; i < AC_MSG_CNT; i++) {
        hcclMsgArea_->sendMsgList[i].valid = ~AC_MSG_VALID_MASK_TAG;
        hcclMsgArea_->commitTurnCnt[i].cnt = 0;
        hcclMsgArea_->finishedTurnCnt[i].cnt = 0;
    }
    hcclMsgArea_->controlMsg.restart = 1;
    #ifdef __aarch64__
    __asm__ __volatile__("dsb st" : : : "memory");
    #endif
}

std::string AicpuRpcServerV2::HcclMsgToStr(HcclMsgV1 *msg)
{
    std::stringstream ss;
    ss << std::to_string((int)msg->commType) << ",";
    ss << std::to_string((int)msg->opType) << ",";
    ss << std::to_string(msg->sendBuffer) << ",";
    ss << std::to_string(msg->recvBuffer) << ",";
    ss << std::to_string(msg->dataCnt) << ",";
    ss << std::to_string(msg->strideCount) << ",";

    ss << std::to_string(msg->ccOpTilingData) << ",";
    ss << std::to_string(msg->valid) << ",";
    ss << std::to_string((int)msg->hcclDataType) << ",";
    ss << std::to_string(msg->repeatCnt) << ",";

    ss << std::to_string((int)msg->selfHandleID) << ",";
    ss << std::to_string((int)msg->seqNum) << ",";
    ss << std::to_string((int)msg->version) << ".";
    ss << std::to_string((int)msg->xorCheck) << ".";
    return ss.str();
}

std::string AicpuRpcServerV2::HcclMsgToTtr(HcclMsg *msg)
{
    if (msg->version == MC2_API_HCCLMSG_VERSION_V1) {
        return HcclMsgToStr(reinterpret_cast<HcclMsgV1 *>(msg));
    }
    std::stringstream ss;
    ss << std::to_string((int)msg->commType) << ",";
    ss << std::to_string((int)msg->opType) << ",";
    ss << std::to_string(msg->sendBuffer) << ",";
    ss << std::to_string(msg->recvBuffer) << ",";
    ss << std::to_string(msg->dataCnt) << ",";
    ss << std::to_string(msg->strideCount) << ",";
    ss << std::to_string((int)msg->hcclDataType) << ",";
    ss << std::to_string(msg->p2pSrcDestRankId) << ",";
    ss << std::to_string(msg->valid) << ",";
    ss << std::to_string((int)msg->repeatCnt) << ",";
    ss << std::to_string((int)msg->everyTurnRsp) << ",";
    ss << std::to_string((int)msg->everyTurnWait) << ",";
    ss << std::to_string((int)msg->commDepGroupID) << ",";
    ss << std::to_string((int)msg->commDepHandleID) << ",";
    ss << std::to_string((int)msg->selfHandleID) << ",";
    ss << std::to_string((int)msg->seqNum) << ",";
    ss << std::to_string((int)msg->version) << ".";
    ss << std::to_string((int)msg->xorCheck) << ".";
    return ss.str();
}

void AicpuRpcServerV2::HcclMsgExtToStr(const HcclMsgExt &msg, u32 rankSize)
{
    std::stringstream ssSendCounts;
    for (uint32_t i = 0; i < rankSize; i++) {
        ssSendCounts << msg.sendCounts[i] << ",";
    }
    std::stringstream ssSendOffset;
    for (uint32_t i = 0; i < rankSize; i++) {
        ssSendOffset << msg.sendOffset[i] << ",";
    }
    std::stringstream ssRecvCounts;
    for (uint32_t i = 0; i < rankSize; i++) {
        ssRecvCounts << msg.recvCounts[i] << ",";
    }
    std::stringstream ssRecvOffset;
    for (uint32_t i = 0; i < rankSize; i++) {
        ssRecvOffset << msg.recvOffset[i] << ",";
    }
    HCCL_ERROR("sendCounts:%s.", ssSendCounts.str().c_str());
    HCCL_ERROR("sendOffset:%s.", ssSendOffset.str().c_str());
    HCCL_ERROR("recvCounts:%s.", ssRecvCounts.str().c_str());
    HCCL_ERROR("sendOffset:%s.", ssSendOffset.str().c_str());
}

void AicpuRpcServerV2::PrintAllHcclMsgArea(u32 rankSize)
{
    HcclMsgArea *hcclMsgArea = reinterpret_cast<HcclMsgArea *>(hcclMsgArea_);
    HCCL_ERROR("********* msgArea %p start print **********", hcclMsgArea);
#ifdef CCL_LLT
    return;
#endif
    for (uint32_t i = 0; i < AC_MSG_CNT; i++) {
        HCCL_ERROR("SendMsg[%u]: %s", i, HcclMsgToTtr(&hcclMsgArea->sendMsgList[i]).c_str());
    }
    for (uint32_t i = 0; i < AC_MSG_CNT; i++) {
        HCCL_ERROR("RecvMsg[%u]: %s", i, HcclMsgToTtr(&hcclMsgArea->recvMsgList[i]).c_str());
    }
    for (uint32_t i = 0; i < AC_MSG_CNT; i++) {
        HCCL_ERROR("HcclMsgExt[%u]:", i);
        HcclMsgExtToStr(hcclMsgArea->paramExtMsgList[i], rankSize);
    }
    std::stringstream ss;
    for (uint32_t i = 0; i < AC_MSG_CNT; i++) {
        ss << hcclMsgArea->commitTurnCnt[i].cnt << ",";
    }
    HCCL_ERROR("CommitTurnCnt:%s", ss.str().c_str());
    ss.str("");
    ss.clear();
    for (uint32_t i = 0; i < AC_MSG_CNT; i++) {
        ss << hcclMsgArea->finishedTurnCnt[i].cnt << ",";
    }
    HCCL_ERROR("FinishedTurnCnt:%s", ss.str().c_str());
    HCCL_ERROR("ControlMsg:%d|%d", hcclMsgArea->controlMsg.restart, hcclMsgArea->controlMsg.restarting);
    HCCL_ERROR("********* msgArea %p end print **********", hcclMsgArea);
}
